package org.study.hadoop.md.ch2;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.Map;

import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.util.Tool;
import org.study.hadoop.md.util.MRDPUtils;

/**
 * 数值概要3： 对于给定的用户评论列表，确定其一天每个小时的评论长度的中位数和标准差。
 */
public class MedianStdDevDriver extends Configured implements Tool {

	public static class MedianStdDevReducer extends Reducer<IntWritable, IntWritable, IntWritable, MedianStdDevTuple> {
		private MedianStdDevTuple result = new MedianStdDevTuple();
		private ArrayList<Float> commentLengths = new ArrayList<Float>();
		
		@Override
		protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context)
				throws IOException, InterruptedException {
			float sum = 0;
			float count = 0;
			commentLengths.clear();
			result.setStddev(0);
			for (IntWritable val : values) {
				commentLengths.add((float) val.get());
				sum += val.get();
				++count;
			}
			Collections.sort(commentLengths);
			if (count % 2 == 0) {
				result.setMedian((commentLengths.get((int) count / 2 - 1) + commentLengths.get((int) count / 2)) / 2.0f);
			} else {
				result.setMedian(commentLengths.get((int) count / 2));
			}
			float mean = sum / count;
			float sumOfSquares = 0.0f;
			for (Float f : commentLengths) {
				sumOfSquares += (f - mean) * (f - mean);
			}
			result.setStddev((float) Math.sqrt(sumOfSquares / (count - 1)));
			context.write(key, result);
		}
	}
	
	public static class MedianStdDevMapper extends Mapper<Object, Text, IntWritable, IntWritable> {
		private IntWritable outHour = new IntWritable();
		private IntWritable outCommentLength = new IntWritable();
		
		private final static SimpleDateFormat FRMT = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
	
		@SuppressWarnings("deprecation")
		@Override
		protected void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			Map<String, String> parsed = MRDPUtils.transformXmlToMap(value.toString());
			String strDate = parsed.get("CreationDate");
			String text = parsed.get("Text");
			if (strDate == null || text == null) {
				return;
			}
			try {
				Date creationDate = FRMT.parse(strDate);
				outHour.set(creationDate.getHours());
				outCommentLength.set(text.length());
				context.write(outHour, outCommentLength);
			} catch (ParseException e) {
				e.printStackTrace();
				return;
			}
		}
	}

	public static class MedianStdDevTuple implements Writable {
		private float median = 0;
		private float stddev = 0;
		
		public float getMedian() {
			return median;
		}
		public void setMedian(float median) {
			this.median = median;
		}
		public float getStddev() {
			return stddev;
		}
		public void setStddev(float stddev) {
			this.stddev = stddev;
		}

		@Override
		public void write(DataOutput out) throws IOException {
			out.writeFloat(median);
			out.writeFloat(stddev);
		}

		@Override
		public void readFields(DataInput in) throws IOException {
			median = in.readFloat();
			stddev = in.readFloat();
		}
		
		@Override
		public String toString() {
			return median + "\t" + stddev;
		}
	}
	
	@Override
	public int run(String[] args) throws Exception {
		return 0;
	}
	
	public static void main(String[] args) {
		
	}
}
